package com.gitee.jastee.kafka.tool;

import cn.hutool.log.Log;
import cn.hutool.log.LogFactory;
import com.gitee.jastee.kafka.producer.KafkaProducerClient;
import com.gitee.jastee.util.ParameterTool;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.concurrent.ExecutionException;

import static com.gitee.jastee.kafka.constant.KafkaConstants.DATA;
import static com.gitee.jastee.kafka.constant.KafkaConstants.KAFKA_TOPIC;

/**
 * @author Jast
 * @description
 * @date 2022-03-03 16:10
 */
public class KafkaProducerTool {

    private static final Log log = LogFactory.get();

    public static void send(String[] args) throws ExecutionException, InterruptedException {
        ParameterTool parameterTool = ParameterTool.fromArgs(args);
        String topicName = parameterTool.getRequired(KAFKA_TOPIC);
        String data = parameterTool.getRequired(DATA);
        KafkaProducerClient kafkaProducerClient = new KafkaProducerClient(parameterTool);
        Producer<String, String> producer = kafkaProducerClient.getProducer();
        producer.send(new ProducerRecord<String, String>(topicName, data)).get();
        log.info("数据发送成功,Topic:{},Data:{}", topicName, data);
    }
}
